DynamoDB Operators & Hooks
Recording Pipeline State Somewhere Durable
DynamoDB shows up in Airflow pipelines less as a "big data destination" and more as a fast, cheap place to record run metadata, idempotency markers, or small lookup/config tables that other tasks read back — this site's own analytics table works exactly this way.
Writing With DynamoDBHook
Unlike most AWS services here, DynamoDB is reached almost entirely through its Hook rather than a dedicated operator — the interactions are usually simple enough (a put_item, a query) that wrapping them in a PythonOperator + Hook is the idiomatic pattern.
from airflow.providers.amazon.aws.hooks.dynamodb import DynamoDBHook
def write_run_metadata(**context):
hook = DynamoDBHook(aws_conn_id="aws_default", table_name="pipeline_runs", table_keys=["run_id"])
table = hook.conn.Table("pipeline_runs")
table.put_item(Item={
"run_id": context["run_id"],
"status": "completed",
"rows_processed": 4821,
"processed_at": context["ts"],
})
Run for real — a Lambda-invocation task's result written straight into a DynamoDB table via this exact pattern:
Figure — write_result_to_dynamodb uses DynamoDBHook(...).conn.Table(...).put_item(...) against a real table, right after the Lambda task above completes.
S3ToDynamoDBOperator — Bulk Loading
For loading a batch of records from S3 rather than writing one item at a time:
from airflow.providers.amazon.aws.transfers.s3_to_dynamodb import S3ToDynamoDBOperator
load_products = S3ToDynamoDBOperator(
task_id="load_products_to_dynamodb",
s3_bucket="my-data-lake",
s3_key="exports/products.json",
dynamodb_table_name="products",
input_format="JSON",
aws_conn_id="aws_default",
)
Reading Back
def get_config_value(key: str, **context):
hook = DynamoDBHook(aws_conn_id="aws_default", table_name="pipeline_config", table_keys=["config_key"])
table = hook.conn.Table("pipeline_config")
response = table.get_item(Key={"config_key": key})
return response.get("Item", {}).get("value")
| Pattern | Method |
|---|---|
| Fetch a single known item | table.get_item(Key={...}) |
| Fetch multiple by partition key | table.query(KeyConditionExpression=...) |
| Scan the whole table (avoid on large tables) | table.scan() |
Airflow Variables and XComs live inside Airflow's own metadata DB, scoped to Airflow. DynamoDB items are visible to anything with the right IAM permissions — a Lambda, another team's service, a dashboard — making it the right choice the moment pipeline state needs to be readable outside Airflow itself.